package com.xiaozhicha.mq.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Before;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Properties;

public class ProduceTest {

    private Properties properties;

    @Before
    public void init(){
        properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    @Test
    public void produce(){
        System.out.println("begin produce");
        try {
            connectionKafka();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("finish produce");
    }

    public void connectionKafka() throws Exception{
        KafkaProducer<String,String> produce = new KafkaProducer(properties);
        int i = 0;
        while (true){
            Thread.sleep(3000);
            produce.send(new ProducerRecord<String, String>("test-topic",++i + " hello"));
        }
    }
}
